//copilot begin
package com.codingfly.config;

import com.codingfly.listener.RocketMQConsumer3;
import jakarta.annotation.PostConstruct;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

import java.util.List;

@Configuration
public class RocketMQConfig {

    private static final Logger logger = LoggerFactory.getLogger(RocketMQConfig.class);

    @Value("${rocketmqCluster1.name-server}")
    public String nameServer;

    @Value("${rocketmqCluster1.producer.group}")
    public String producerGroup;

    @Value("${rocketmqCluster1.producer.send-message-timeout}")
    public Integer sendMsgTimeout;

    @Value("${rocketmqCluster1.producer.max-message-size}")
    public Integer maxMessageSize;

    @Value("${rocketmqCluster1.producer.retry-times-when-send-failed}")
    public Integer retryTimesWhenSendFailed;

    @Value("${rocketmqCluster1.producer.retry-times-when-send-async-failed}")
    public Integer retryTimesWhenSendAsyncFailed;

    @Value("${rocketmqCluster1.consumer1.group}")
    public String consumer1Group;

    @Value("${rocketmqCluster1.consumer1.topic}")
    public String consumer1Topic;

    @Value("${rocketmqCluster1.consumer1.pull-batch-size}")
    public Integer consumer1PullBatchSize;

    @Value("${rocketmqCluster1.consumer2.group}")
    public String consumer2Group;

    @Value("${rocketmqCluster1.consumer2.topic}")
    public String consumer2Topic;

    @Value("${rocketmqCluster1.consumer2.pull-batch-size}")
    public Integer consumer2PullBatchSize;

    @Bean
    @Primary
    public RocketMQTemplate rocketMQTemplate(@Qualifier(value = "producer") DefaultMQProducer producer) {
        RocketMQTemplate rocketMqTemplate = new RocketMQTemplate();
        rocketMqTemplate.setProducer(producer);
        return rocketMqTemplate;
    }

    @Primary
    @Bean("producer")
    public DefaultMQProducer defaultMqProducer() {
        DefaultMQProducer producer = new DefaultMQProducer();
        producer.setNamesrvAddr(this.nameServer);
        producer.setProducerGroup(this.producerGroup);
        producer.setSendMsgTimeout(this.sendMsgTimeout);
        producer.setMaxMessageSize(this.maxMessageSize);
        producer.setRetryTimesWhenSendFailed(this.retryTimesWhenSendFailed);
        producer.setRetryTimesWhenSendAsyncFailed(this.retryTimesWhenSendAsyncFailed);
        return producer;
    }

    // 初始化完一个PullConsumer后, 当前线程就while(true)接收信息, 代码跳不出while(true), 导致该线程后面的代码无法执行, 无法初始化其他PullConsumer
    @PostConstruct
    public void init() throws MQClientException {
        new Thread() {
            @Override
            public void run() {
                try {
                    initPullConsumer1();
                    logger.info("初始化PullConsumer1结束");
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
            }
        }.start();

        new Thread() {
            @Override
            public void run() {
                try {
                    initPullConsumer2();
                    logger.info("初始化PullConsumer2结束");
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
            }
        }.start();
    }

    public void initPullConsumer1() throws MQClientException {
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer();
        consumer.setNamesrvAddr(nameServer);
        consumer.setConsumerGroup(consumer1Group);
        consumer.setPullBatchSize(consumer1PullBatchSize);
        consumer.subscribe(consumer1Topic, "*");  // 订阅Topic, 可指定Tag进行消息过滤, *表示订阅所有Tag
        consumer.start(); // 启动消费者
        try {
            while (true) {  // 4. 循环拉取消息（推荐使用while(true)持续消费）
                List<MessageExt> messages = consumer.poll(5000); // 拉取消息，设置超时时间为5秒
                if (messages != null && !messages.isEmpty()) {
                    for (MessageExt msg : messages) { // 5. 处理消息
                        System.out.printf("收到消息: 主题=%s, 标签=%s, 消息ID=%s, 内容=%s%n",
                                msg.getTopic(), msg.getTags(), msg.getMsgId(), new String(msg.getBody()));
                        processMessage(msg); // 模拟业务处理
                        RocketMQConsumer3.printCostTime(new String(msg.getBody()));
                    }
                    consumer.commitSync(); // 6. 手动提交消费偏移量（默认模式：AUTO_COMMIT）
                } else {
                    Thread.sleep(100); // 没有消息，可短暂休眠避免CPU空转
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            consumer.shutdown(); // 7. 关闭消费者
        }
    }

    public void initPullConsumer2() throws MQClientException {
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer();
        consumer.setNamesrvAddr(nameServer);
        consumer.setConsumerGroup(consumer2Group);
        consumer.setPullBatchSize(consumer2PullBatchSize);
        consumer.subscribe(consumer2Topic, "*");  // 订阅Topic, 可指定Tag进行消息过滤, *表示订阅所有Tag
        consumer.start(); // 启动消费者
        try {
            while (true) {  // 4. 循环拉取消息（推荐使用while(true)持续消费）
                List<MessageExt> messages = consumer.poll(5000); // 拉取消息，设置超时时间为5秒
                if (messages != null && !messages.isEmpty()) {
                    for (MessageExt msg : messages) { // 5. 处理消息
                        System.out.printf("收到消息: 主题=%s, 标签=%s, 消息ID=%s, 内容=%s%n",
                                msg.getTopic(), msg.getTags(), msg.getMsgId(), new String(msg.getBody()));
                        processMessage(msg); // 模拟业务处理
                        RocketMQConsumer3.printCostTime(new String(msg.getBody()));
                    }
                    consumer.commitSync(); // 6. 手动提交消费偏移量（默认模式：AUTO_COMMIT）
                } else {
                    Thread.sleep(100); // 没有消息，可短暂休眠避免CPU空转
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        } finally {
            consumer.shutdown(); // 7. 关闭消费者
        }
    }

    private void processMessage(MessageExt message) {
        // 处理消息的业务逻辑
    }

}
//copilot end